
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include <fcntl.h>   // 对于 fcntl 和 O_NONBLOCK
#include <unistd.h>  // 对于 read 函数

#define MAX_EVENTS 10000
#define PORT 8080
#define THREAD_POOL_SIZE 4
#define CHECK_INTERVAL 5 // 检查间隔，单位：秒
#define THREAD_POOL_SIZE 4

// 任务结构体
typedef struct task {
    void (*function)(void*);
    void *argument;
    struct task *next;
} task_t;

// 线程池结构体
typedef struct thread_pool {
    pthread_t threads[THREAD_POOL_SIZE];
    task_t *task_queue_head;
    task_t *task_queue_tail;
    pthread_mutex_t lock;
    pthread_cond_t notify;
    int stop;
} thread_pool_t;

// 线程池全局变量
thread_pool_t pool;

// 线程函数
void* thread_do(void* arg) {
    while (1) {
        task_t *task;

        pthread_mutex_lock(&pool.lock);
        while (pool.task_queue_head == NULL && !pool.stop) {
            pthread_cond_wait(&pool.notify, &pool.lock);
        }
        if (pool.stop) {
            break;
        }
        task = pool.task_queue_head;
        pool.task_queue_head = task->next;
        pthread_mutex_unlock(&pool.lock);

        (*(task->function))(task->argument);
        free(task);
    }
    pthread_mutex_unlock(&pool.lock);
    pthread_exit(NULL);
    return NULL;
}

// 初始化线程池
int thread_pool_init(thread_pool_t *pool) {
    pool->task_queue_head = NULL;
    pool->task_queue_tail = NULL;
    pool->stop = 0;
    pthread_mutex_init(&(pool->lock), NULL);
    pthread_cond_init(&(pool->notify), NULL);

    for (int i = 0; i < THREAD_POOL_SIZE; i++) {
        if (pthread_create(&(pool->threads[i]), NULL, thread_do, NULL) != 0) {
            return -1;
        }
    }
    return 0;
}


// 销毁线程池
int thread_pool_destroy(thread_pool_t *pool) {
    pool->stop = 1;
    pthread_cond_broadcast(&(pool->notify));

    for (int i = 0; i < THREAD_POOL_SIZE; i++) {
        pthread_join(pool->threads[i], NULL);
    }
    pthread_mutex_destroy(&(pool->lock));
    pthread_cond_destroy(&(pool->notify));
    return 0;
}
// 提交任务到线程池
void thread_pool_submit(void (*function)(void*), void* argument) {
    task_t *new_task = (task_t*)malloc(sizeof(task_t));
    new_task->function = function;
    new_task->argument = argument;
    new_task->next = NULL;

    pthread_mutex_lock(&pool.lock);
    if (pool.task_queue_head == NULL) {
        pool.task_queue_head = new_task;
        pool.task_queue_tail = new_task;
    } else {
        pool.task_queue_tail->next = new_task;
        pool.task_queue_tail = new_task;
    }
    pthread_cond_signal(&pool.notify);
    pthread_mutex_unlock(&pool.lock);
}

// 全局列表和互斥锁来存储和管理活动的socket
int active_sockets[MAX_EVENTS];
pthread_mutex_t sockets_mutex = PTHREAD_MUTEX_INITIALIZER;

// 假设你有一个线程池的实现
// 这里只是一个用于演示的函数原型
// void thread_pool_submit(void (*function)(void*), void* argument);

void handle_connection(int sockfd) {
    // 在这里处理连接
    char buffer[1024];
    ssize_t bytes_read;

    bytes_read = read(sockfd, buffer, sizeof(buffer) - 1);
    if (bytes_read > 0) {
        buffer[bytes_read] = '\0';
        printf("Received: %s\n", buffer);
        write(sockfd, buffer, bytes_read); // Echo back
    }

    // 执行关闭前的业务逻辑
    // ...

    // close(sockfd);
}

void* worker_function(void* arg) {
    int sockfd = *(int*)arg;
    free(arg);
    handle_connection(sockfd);
    return NULL;
}



void* timer_thread_function(void* arg) {
    while (1) {
        sleep(CHECK_INTERVAL);
        pthread_mutex_lock(&sockets_mutex);

        printf("kill a conect\n");

        // 检查和关闭连接的逻辑
        // ...
        pthread_mutex_unlock(&sockets_mutex);
    }
    return NULL;
}

int main() {

    memset(active_sockets, -1, sizeof(active_sockets)); // 初始化socket列表

    // 创建定时器线程
    // todo 暂时关闭检查
    // pthread_t timer_thread;
    // if (pthread_create(&timer_thread, NULL, timer_thread_function, NULL) != 0) {
    //     perror("Failed to create the timer thread");
    //     return 1;
    // }
    // pthread_detach(timer_thread); // 不需要join


    if (thread_pool_init(&pool) != 0) {
        fprintf(stderr, "Failed to create thread pool.\n");
        return 1;
    }


    int server_fd, new_socket, epoll_fd;
    struct epoll_event event, events[MAX_EVENTS];
    struct sockaddr_in address;
    int addrlen = sizeof(address);

    // 创建socket
    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }

    // 绑定socket到端口
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(PORT);

    if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }

    // 监听socket
    if (listen(server_fd, 10) < 0) {
        perror("listen");
        exit(EXIT_FAILURE);
    }

    // 创建epoll实例
    if ((epoll_fd = epoll_create1(0)) == -1) {
        perror("epoll_create1");
        exit(EXIT_FAILURE);
    }

    event.events = EPOLLIN;
    event.data.fd = server_fd;
    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &event) == -1) {
        perror("epoll_ctl");
        exit(EXIT_FAILURE);
    }

    // 事件循环
    while (1) {
        int n, i;

        n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        for (i = 0; i < n; i++) {
            if (events[i].events & EPOLLRDHUP) {
                // 对端关闭了连接
                printf("have a conect close ------\n");
                close(events[i].data.fd);

                pthread_mutex_lock(&sockets_mutex);
                // 从活动socket列表中移除
                for (int j = 0; j < MAX_EVENTS; ++j) {
                    if (active_sockets[j] == events[i].data.fd) {
                        active_sockets[j] = -1;
                        break;
                    }
                }
                pthread_mutex_unlock(&sockets_mutex);
            } else if (events[i].data.fd == server_fd) {
                printf("epoll wait somethings\n");
                // 接受连接
                new_socket = accept(server_fd, (struct sockaddr*)&address, (socklen_t*)&addrlen);

                if (new_socket < 0) {
                    perror("accept");
                    exit(EXIT_FAILURE);
                }

                // 设置为非阻塞模式
                int flags = fcntl(new_socket, F_GETFL, 0);
                fcntl(new_socket, F_SETFL, flags | O_NONBLOCK);

                // 将新连接的socket添加到epoll的监控中
                event.events = EPOLLIN | EPOLLET; // 边缘触发，并添加EPOLLRDHUP事件
                event.data.fd = new_socket;
                if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_socket, &event) == -1) {
                    perror("epoll_ctl: new_socket");
                    exit(EXIT_FAILURE);
                }

                printf("get a conect\n");
                if (new_socket >= 0) {
                    pthread_mutex_lock(&sockets_mutex);
                    for (int i = 0; i < MAX_EVENTS; ++i) {
                        if (active_sockets[i] == -1) {
                            active_sockets[i] = new_socket;
                            break;
                        }
                    }
                    pthread_mutex_unlock(&sockets_mutex);
                }


                // 将新连接的socket分配给线程池
                int* sockfd = malloc(sizeof(int));
                *sockfd = new_socket;
                thread_pool_submit(worker_function, sockfd);
            }
        }
    }


    // 在结束时销毁线程池
    thread_pool_destroy(&pool);

    close(server_fd);
    close(epoll_fd);
    return 0;
}

// 你需要实现thread_pool_submit函数，将任务提交到你的线程池
// void thread_pool_submit(void (*function)(void*), void* argument) {
//
//     // 这里应该将function和argument提交到线程池
//     // 下面的代码仅为示例，实际应用中需要线程池的具体实现
//     pthread_t thread;
//     pthread_create(&thread, NULL, function, argument);
//     pthread_detach(thread);
// }
